﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net;
using System.Runtime.Serialization;
using System.ServiceModel;
using System.Threading;


namespace Paxos
{
    public class Counter
    {
        public long Total = 0;
        public long FirstSuccess = 0;
        public long FirstNoConsensus = 0;
        public long FirstOutOfDate = 0;
        public long SecondSuccess = 0;
        public long SecondNoConsensus = 0;
        public long SecondOutOfDate = 0;

        public override string ToString()
        {
            return String.Format("Total {0} F.Success {1} F.NoConsensus {2} F.OutOfDate {3} S.Success {4} S.NoConsensus {5} S.OutOfDate {6}",
                Total, FirstSuccess, FirstNoConsensus, FirstOutOfDate, SecondSuccess, SecondNoConsensus, SecondOutOfDate);
        }
    }

    public abstract class ReplicaStateMachine
    {
        internal ChannelFactory<IPaxosProproser> _proposerFactory;
        internal ChannelFactory<IPaxosAcceptor> _acceptorFactory;
        internal IEnumerable<IPaxosAcceptor> _acceptors;
        internal IEnumerable<IPaxosProproser> _proposers;
        internal NetTcpBinding _commonBinding;

        internal IPaxosProproser _proposerEP;
        internal IPaxosAcceptor _acceptorEP;
        private ServiceHost _proposerHost;
        private ServiceHost _acceptorHost;

        internal Proposer _localProposer;
        internal Acceptor _localAcceptor;

        internal CommandLogger _commandLogger;

        internal SGUID _id;
        internal ServerView _view;
        internal bool _needToCacheEndpoint;

        public Counter Counter {get; private set;}

        public void Initilize(SGUID id, ServerView view)
        {
            _id = id;
            _view = view;
            Counter = new Paxos.Counter();

            _commonBinding = new NetTcpBinding();
            //_commonBinding.OpenTimeout = TimeSpan.FromMilliseconds(200);
            //_commonBinding.SendTimeout = TimeSpan.FromMilliseconds(500);
            //_commonBinding.ReceiveTimeout = TimeSpan.FromMilliseconds(200);
            //_commonBinding.CloseTimeout = TimeSpan.FromMilliseconds(500);
            
            _localProposer = new Proposer(this);
            _localAcceptor = new Acceptor(String.Format("acceptor.{0}.xml", _id), this);
            _proposerHost = new ServiceHost(_localProposer);
            _proposerHost.AddServiceEndpoint(typeof(IPaxosProproser), _commonBinding, _id.GetProposerEndpointAddress());

            // add acceptor host
            _acceptorHost = new ServiceHost(_localAcceptor);
            _acceptorHost.AddServiceEndpoint(typeof(IPaxosAcceptor), _commonBinding, _id.GetAcceptorEndpointAddress());

            Console.WriteLine("proposer/acceptor endpoint {0} added", _id);

            _proposerFactory = new ChannelFactory<IPaxosProproser>(_commonBinding);
            _acceptorFactory = new ChannelFactory<IPaxosAcceptor>(_commonBinding);            

            // try to replay commands
            _commandLogger = new CommandLogger(String.Format("cmd.{0}.bin", _id));
            _commandLogger.Replay(this);

            CacheEndpoints();

            // catch up
            _localAcceptor.CatchUp();

            // open proposer and acceptor agents.
            _proposerHost.Open();
            _acceptorHost.Open();
        }

        public void LogCommand(PaxosCommand command)
        {
            _commandLogger.Log(command);
        }

        public void Shutdown()
        {
            if (_proposerHost != null)
            {
                _proposerHost.Abort();
                _proposerHost.Close();
            }

            if (_acceptorHost != null)
            {
                _acceptorHost.Abort();
                _acceptorHost.Close();
            }
        }

        public bool Propose(PaxosCommand command)
        {
            // check endpoint
            if (_needToCacheEndpoint)
            {
                CacheEndpoints();
            }

            IssueResult result= _localProposer.IssueCommand(command);
            Console.WriteLine("First issue got {0}", result);

            Interlocked.Increment(ref Counter.Total);

            if (result == IssueResult.Success)
            {
                Interlocked.Increment(ref Counter.FirstSuccess);
                return true;
            }
            else if (result == IssueResult.NoConsensus)
            {
                Interlocked.Increment(ref Counter.FirstNoConsensus);
                return false;
            }
            else //IssueResult.OutOfDate
            {
                Interlocked.Increment(ref Counter.FirstOutOfDate);

                _acceptorEP.CatchUp();
                result = _localProposer.IssueCommand(command);
                Console.WriteLine("Issue after catchup got {0}", result);
                if (result == IssueResult.Success)
                {
                    Interlocked.Increment(ref Counter.SecondSuccess);
                    return true;
                }
                else if (result == IssueResult.NoConsensus)
                {
                    Interlocked.Increment(ref Counter.SecondNoConsensus);
                }
                else if (result == IssueResult.OutOfDate)
                {
                    Interlocked.Increment(ref Counter.SecondOutOfDate);
                }
            }

            return false;
        }

        private void CacheEndpoints()
        {
            if (_acceptorFactory == null || _proposerFactory == null ||  _view == null 
                || _view.Servers == null || _view.Servers.Count == 0)
            {
                return;
            }

            //_proposerEP = _proposerFactory.CreateChannel(new EndpointAddress(_id.GetProposerEndpointAddress()));
            //_acceptorEP = _acceptorFactory.CreateChannel(new EndpointAddress(_id.GetAcceptorEndpointAddress()));
            List<IPaxosProproser> psrs = new List<IPaxosProproser>();
            List<IPaxosAcceptor> acts = new List<IPaxosAcceptor>();
            foreach (SGUID sg in _view.Servers)
            {
                IPaxosProproser p = _proposerFactory.CreateChannel(new EndpointAddress(sg.GetProposerEndpointAddress()));
                IPaxosAcceptor a = _acceptorFactory.CreateChannel(new EndpointAddress(sg.GetAcceptorEndpointAddress()));
                if (_id.CompareTo(sg)==0)
                {
                    _proposerEP = p;
                    _acceptorEP = a;
                }
                psrs.Add(p);
                acts.Add(a);
            }

            _proposers = psrs;
            _acceptors = acts;
            _needToCacheEndpoint = false;
        }
    }
}
